
from API.coingecko import CoinGeckoAPI
from tools.logger import logger
import pandas as pd
import datetime, time
import os,re
import schedule
from tools.task_monitor import send_error_to_email
import traceback
import multiprocessing
import random


def clean_filename(filename):
    # 替换Windows文件名中不允许的字符
    # 这里我们将冒号、问号等替换为下划线
    # 定义不允许的字符集合： \ / : * ? " < > |
    cleaned = re.sub(r'[\\/*?:"<>|]', '_', filename)
    return cleaned

def reload_cex_spot_amount():
    try:
        cg = CoinGeckoAPI()
        days = 365
        spot_cex = ['binance', 'gdax', 'okex', 'huobi', 'gate', 'kucoin', 'crypto_com', 'bitfinex', 'bybit_spot']
        results = []
        for exchange_id in spot_cex:
            logger.info(f'开始获取{exchange_id}现货交易额数据')
            amount = cg.get_exchanges_volume_chart_by_id_v2(id=exchange_id, days=days)
            amount.name = exchange_id
            results.append(amount)
            logger.info(f'{exchange_id}现货交易额数据获取成功')
        df = pd.concat(results, axis=1)
        df.index = df.index.astype(str)
        df.to_csv(os.path.join("E:/data/amount_data_v3/", clean_filename(f'cex_spot_amount_coingecko_reload_{datetime.datetime.now()}.csv')))
        logger.info("reload_cex_spot_amount 执行成功")
    except Exception as e:
        msg = traceback.format_exc()
        logger.info(msg)
        send_error_to_email(script_name=os.path.basename(__file__), func_name="reload_cex_spot_amount", message=msg)


def reload_cex_deri_amount():
    try:
        cg = CoinGeckoAPI()
        days = 365
        deri_cex = ['binance', 'okex', 'bybit', 'kucoin', 'crypto_com', 'huobi', 'gate', 'bitfinex', 'kraken', 'bitmex']
        deri_cex_dict = {'binance': 466, 'okex': 379, 'bybit': 460, 'ftx': 443, 'kucoin': 471, 'bitmex': 378,
                                    'crypto_com': 669, 'huobi': 433, 'gate': 403, 'bitfinex': 486, 'kraken': 426}
        results = []
        for exchange_id in deri_cex:
            logger.info(f'开始获取{exchange_id}现货交易额数据')
            amount = cg.get_exchanges_volume_chart_by_id_v2(id=deri_cex_dict[exchange_id], days=days)
            amount.name = exchange_id
            results.append(amount)
            logger.info(f'{exchange_id}现货交易额数据获取成功')
        df = pd.concat(results, axis=1)
        df.index = df.index.astype(str)
        df.to_csv(os.path.join("E:/data/amount_data_v3/", clean_filename(f'cex_deri_amount_coingecko_reload_{datetime.datetime.now()}.csv')))
        logger.info(f'reload_cex_deri_amount 执行成功')
    except Exception as e:
        msg = traceback.format_exc()
        logger.info(msg)
        send_error_to_email(script_name=os.path.basename(__file__), func_name="reload_cex_deri_amount", message=msg)


def reload_one_dex_amount(cg, exchange_id, days):
    logger.info(f'开始获取dex_{exchange_id}现货交易额数据')
    while True:
        try:
            amount = cg.get_exchanges_volume_chart_by_id_v2(id=exchange_id, days=days)
            break
        except Exception as e:
            logger.info(f"发生异常:{e}")
            time.sleep(10)
    amount.name = exchange_id
    logger.info(f'dex_{exchange_id}现货交易额数据获取成功')
    return amount


def reload_dex_amount():

    try:
        cg = CoinGeckoAPI()
        days = 365
        dex = ['uniswap-bsc', 'uniswap_v2', 'uniswap_v3', 'uniswap_v3_arbitrum', 'uniswap-v3-avalanche',
             'uniswap_v3_optimism', 'uniswap_v3_polygon_pos',
             'curve_arbitrum', 'curve_ethereum', 'curve_optimism',
             'apex_pro', 'maverick_protocol', 'ferro-protocol', 'openocean_finance', 'jupiter', 'orca', 'thorwallet',
             'thorswap', 'woofi', 'pulsex', 'syncswap',
             'balancer-v1', 'balancer-v2', 'balancer_arbitrum',
             'sushiswap', 'sun_io',
             'pancakeswap_aptos', 'pancakeswap_ethereum', 'pancakeswap_new', 'pancakeswap_stableswap', 'pancakeswap-v3-bsc',
             'pancakeswap-v3-ethereum',
             'dodo', 'dodo_arbitrum', 'dodo_bsc', 'dodo_polygon',
             'traderjoe', 'traderjoe-v2-1-arbitrum', 'traderjoe-v2-1-avalanche', 'traderjoe-v2-1-bsc',
             'traderjoe-v2-arbitrum', 'traderjoe-v2-avalanche', 'traderjoe-v2-bsc',
             'quickswap', 'quickswap-polygon-zkevm', 'quickswap_v3',
             ]
        params = []
        flag = random.choice([1, -1])
        for exchange_id in dex[::flag]:
            params.append((cg, exchange_id, days))
        with multiprocessing.Pool(60) as pool:
            results = pool.starmap(reload_one_dex_amount, params)
        df = pd.concat(results, axis=1)
        df.index = df.index.astype(str)

        df.to_csv(os.path.join("E:/data/amount_data_v3/", clean_filename(f'dex_spot_amount_coingecko_reload_{datetime.datetime.now()}.csv')))
        logger.info(f'reload_dex_amount 执行成功')
    except Exception as e:
        msg = traceback.format_exc()
        logger.info(msg)
        send_error_to_email(script_name=os.path.basename(__file__), func_name="reload_dex_amount", message=msg)





if __name__ == "__main__":
    # 设置定时任务
    start_time = "08:00"
    schedule.every().day.at(start_time).do(reload_cex_spot_amount)
    schedule.every().day.at(start_time).do(reload_cex_deri_amount)
    schedule.every().day.at(start_time).do(reload_dex_amount)
    while True:
        schedule.run_pending()
        time.sleep(1)  # 每秒检查一次
